package com.epidemy.actors

import scala.actors.Actor
import scala.actors.remote.RemoteActor
import scala.actors.remote.Node
import scala.collection.mutable.HashMap
import java.net.InetAddress

import scala.collection.JavaConversions._
import com.epidemy.Grid
import com.epidemy.jmdns.DNSActor
import javax.jmdns.{JmDNS, ServiceInfo}
import com.epidemy.Grid._
import java.util.{Random, Date}
import java.io._
import org.teleal.cling.UpnpServiceImpl
import org.teleal.cling.support.igd.{PortMappingListener, PortMapping}

object NodeActor {
  val random = new Random(System.currentTimeMillis)
  def createUniqueName(nBytes: Int) = {
    val bytes = new Array[Byte](nBytes)
    random.nextBytes(bytes)
    bytes.map(b => Integer.toHexString(b)).mkString("")
  }

  val epidemyNodeKey = "_epidemynode._tcp.local."

  case class GetTopology(includeThis: Boolean)
}

class NodeActor(val thisNode: GridNode)
extends Actor
   with ProcessTracker
{
  import NodeActor._
  protected val topology = new HashMap[GridNode, Date]()

  //protected def topologyWithThis = (topology ++ Map(thisNode -> now)).toList

  protected def now = new Date(System.currentTimeMillis)

  protected def ponged(remoteNode: GridNode) =
    if (remoteNode != thisNode)
      topology(remoteNode) = now
  
  protected def newTopologyUpdate(requestReply: Boolean) =
    TopologyUpdate(thisNode, Topology(topology.toList), requestReply)

  protected def remote(node: GridNode) =
    RemoteActor.select(node.node, node.localName)

  def err(info: ServiceInfo) =
    println("ill-formed service info : " + info)
  
  def getNode(info: ServiceInfo) =
    try {
      val server = info.getPropertyString("server")
      val port = info.getPropertyString("port").trim
      var localName = info.getPropertyString("localName")
      while (localName.startsWith("'"))
        localName = localName.substring(1)
      Some(GridNode(Node(server, Integer.parseInt(port)), scala.Symbol(localName)))
    } catch { case ex =>
      ex.printStackTrace
      None
    }

  def prefix = "[" + thisNode + "] "

  def pingAll(topology: Traversable[GridNode]) = {
    //println(prefix + "Pinging " + topology.size + " nodes : " + topology.map(_._1.localName.name).mkString(", "))
    val ping = Ping(thisNode)
    topology.map(n => {
      if (n != thisNode) {
        val a = remote(n)
        //println("pinging " + n)
        a ! ping
      }
    })
  }
  
  private def hashtable(map: Map[String, String]) = {
    val jmap: java.util.Map[String, String] = map
    val ht: java.util.Hashtable[String, String] = new java.util.Hashtable(map.size)
    ht.putAll(map)
    ht
  }

  /*
  def openUpnpPort = {
    val desiredMapping =
    new PortMapping(
      thisNode.port,//8123,
      thisNode.address,//"192.168.0.123",
      PortMapping.Protocol.TCP,
      "Epidemy Port Mapping for " + thisNode
    )
    val upnpService =
      new UpnpServiceImpl(
        new PortMappingListener(desiredMapping)
      );

    upnpService.getControlPoint.search
  }
  */
  
  override def act: Unit = {
    thisNode

    val jmdns = JmDNS.create
    jmdns.registerService(ServiceInfo.create(
      epidemyNodeKey,
      createUniqueName(8),
      thisNode.node.port,
      0,
      0,
      hashtable(Map(
        "server" -> thisNode.node.address,
        "port" -> thisNode.node.port.toString,
        "localName" -> thisNode.localName.name
      ))
    ))
    new DNSActor(
      jmdns,
      epidemyNodeKey,
      resolved = info => getNode(info) match {
        case Some(node) =>
          if (node != thisNode)
            NodeActor.this ! ConnectTo(node)
        case None =>
          err(info)
      },
      unresolved = info => getNode(info) match {
        case Some(node) =>
          if (node != thisNode)
            NodeActor.this ! DisconnectFrom(node)
        case None =>
          err(info)
      }
    ).start

    {
      import Ticker._
      Ticker.start
      Ticker ! new Every(15 seconds, 'ShouldPingEveryone)
    }
    RemoteActor.alive(thisNode.node.port)
    RemoteActor.register(thisNode.localName, this)

    println("Grid node active : " + thisNode.describe)
    loop {
      react {
        case 'ShouldPingEveryone =>
          pingAll(topology.keys)

        case Ping(sourceNode: GridNode) =>
          println(prefix + "Received Ping from " + sourceNode)
          sender ! Pong(thisNode)

        case Pong(sourceNode: GridNode) =>
          println(prefix + "Received Pong from " + sourceNode)
          ponged(sourceNode)

        case GetTopology(includeThis) =>
          reply {
            Topology({
              val others = topology.toList
              if (includeThis)
                (thisNode, now) :: others
              else
                others
            })
          }
        
        case TopologyUpdate(sourceNode, Topology(topology), requestReply) =>
          println(prefix + "Received topology update from " + sourceNode + " : " + topology.mkString(", "))
          //updateTopology(Seq(sourceNode -> new Date(System.currentTimeMillis)))
          ponged(sourceNode)
          val thisTop = this.topology
          pingAll(topology.map(_._1).filter(n => !thisTop.containsKey(n)))
          if (requestReply)
            sender ! newTopologyUpdate(false)

        case RequestTopologyUpdate =>
          sender ! newTopologyUpdate(false)

        case ConnectTo(node) =>
          println(prefix + "Connecting to " + node)
          remote(node) ! Ping(thisNode)
          remote(node) ! newTopologyUpdate(true)

        case DisconnectFrom(node) =>
          //println(prefix + "Disconnecting from " + node)
          topology.remove(node)

        case RunProgram(sourceNode, Program(code, url)) =>
          // http://download.oracle.com/javase/1.4.2/docs/guide/security/spec/security-spec.doc6.html
          val privileged = true

          val ext = ".o"
          val f = File.createTempFile("tmp-" + sourceNode.node.address + "-" + sourceNode.localName.name + "-", ext)
          val workingDir = new File({ val fs = f.toString; fs.substring(0, fs.length - ext.length) })
          workingDir.mkdirs
          val out = new FileOutputStream(f)
          out.write(code)
          out.close
          val permFile = new File(f + ".policy")
          if (privileged) {
            val permOut = new PrintStream(permFile)
            permOut.println("""
grant {
  permission java.net.SecurityException
        """ + "\"" + url.getHost + ":" + url.getPort + "\"" + """, "connect";
};
""")
            permOut.close
          }
          val pb = new ProcessBuilder(
            "java",
            "-cp",
            System.getProperty("java.class.path"),
            "-Djava.security.policy=" + permFile, // http://java.sun.com/developer/onlineTraining/Programming/JDCBook/signed2.html
            "com.epidemy.ProgramRunner",
            privileged.toString,
            f.getAbsolutePath,
            url.toString
          )
          pb.directory(workingDir)
          pb.redirectErrorStream
          val p = pb.start
          val id = addProcess(p)
          scala.concurrent.ops.spawn {
            try {
              val in = new BufferedReader(new InputStreamReader(p.getInputStream))
              val prePrefix = "[" + f.getName
              val prefix = prePrefix + "] "
              var line = ""
              while ({ line = in.readLine; line != null })
                println(prefix + line)
              println(prePrefix + " Exit = " + p.waitFor + "]")
            } finally {
              f.delete
              if (privileged)
                permFile.delete
              val workingFiles = workingDir.listFiles
              if (workingFiles != null) {
                if (workingFiles.length > 0) {
                  println("Process left " + workingFiles.size + " files in its working directory :")
                  workingFiles.foreach(wf => println("\t" + wf))
                } else
                  workingDir.delete
              }
              removeProcess(id)
            }
          }
          RemoteProcess(thisNode, id)

        case KillProcess(id) =>
          val p = removeProcess(id)
          p.destroy
        
        case msg =>
          println(prefix + "Received unexpected message : " + msg)
      }
    }
  }
}

trait ProcessTracker {

  private var nextProcessId = 1L
  private var processById = Map[Long, Process]()
  //private var idByProcess = Map[Process, Long]()
  private lazy val registerHook: Unit = {
    Runtime.getRuntime.addShutdownHook(new Thread {
      override def run = this synchronized {
        println("Forcibly killing " + processById.size + " child processes")
        processById.values.map(p => try { p.destroy } catch { case ex => ex.printStackTrace })
      }
    })
  }
  protected def addProcess(process: Process): Long = this synchronized {
    registerHook
    val id = nextProcessId
    nextProcessId += 1
    processById += ((id, process))
    //idByProcess += ((process, id))
    id
  }
  protected def removeProcess(id: Long): Process = this synchronized {
    val p = processById(id)
    processById -= id
    //idByProcess -= p
    p
  }
}